package com.udf.common;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

// 定义累加器，保存每个玩家的当前总和和是否已触发
class SumAccumulator {
    public double total = 0.0;
    public boolean alertTriggered = false;
}

@FunctionHint(
    output = @DataTypeHint("ROW<player_id INT, alert_message STRING>")
)
public class CumulativeBetAlert extends AggregateFunction<Row, SumAccumulator> {

    @Override
    public SumAccumulator createAccumulator() {
        return new SumAccumulator();
    }

    // 累加逻辑：更新总和，检查是否触发告警
    public void accumulate(SumAccumulator acc, Integer playerId, Double wager) {
        acc.total += wager;
        if (acc.total >= 8000 && !acc.alertTriggered) {
            acc.alertTriggered = true;
        }
    }

    // 返回告警信息（仅在首次达到阈值时触发）
    @Override
    public Row getValue(SumAccumulator acc) {
        if (acc.alertTriggered) {
            return Row.of(
                // 假设 player_id 通过参数传递，此处需调整逻辑
                "玩家累计投注已超过 8000，当前总额：" + acc.total
            );
        }
        return null; // 不触发时返回 null
    }
}